package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_02_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeCartAdd().init(
                2001,
                2,
                "Dwd_02_DwdTradeCartAdd"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1.读取 ods 数据
        readOdsDb(tEnv, "Dwd_02_DwdTradeCartAdd");


        // 2. 从 ods_db中过滤出加购
        Table cartInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id," +
                " `data`['user_id'] user_id," +
                " `data`['course_id'] course_id," +
                " `data`['cart_price'] cart_price," +
                " ts ," +
                " pt " +
                " from ods_db" +
                " where `database`='gmall2022'" +
                " and `table`='cart_info'" +
                " and `type`='insert'");

        // 3. 写入到kafka中
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                " id string," +
                " user_id string," +
                " course_id string," +
                " cart_price string," +
                " ts bigint)"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD));
        cartInfo.executeInsert("dwd_trade_cart_add");

    }
}
